package com.simple.datax.conn.impl;

import java.util.NoSuchElementException;

import org.apache.commons.pool.KeyedObjectPool;
import org.apache.commons.pool.impl.GenericKeyedObjectPool;
import org.apache.commons.pool.impl.GenericKeyedObjectPoolFactory;

import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.simple.datax.conn.BaseConnector;
import com.simple.datax.conn.ConnectorPool;
import com.simple.datax.core.model.connpara.CommonConnPara;

/** 
 * @Title: ReceiveConnectorPool.java 
 * @Package com.simple.datax.conn.impl 
 * @Description: This class define connector pool for receiving messages. All connector
 * in this pool has only one instance, so only one thread can get it at one time.
 * @author Joe Jin 
 * @date 2012-12-25 
 * @version V1.0 
 */
public class ReceiveConnectorPool extends ConnectorPool {
	private static final Logger logger = LoggerFactory
			.getLogger(ReceiveConnectorPool.class);
	private ReceiveConnectorPool() {
		
	}
	private static KeyedObjectPool<String, BaseConnector> internal_pool = null;
	private static ReceiveConnectorPool pool = new ReceiveConnectorPool();

	/**
	 * Singleton
	 * @return
	 */
	public static synchronized ReceiveConnectorPool getInstance() {
		if (internal_pool == null) {
			GenericKeyedObjectPoolFactory<String, BaseConnector> factory = new GenericKeyedObjectPoolFactory<String, BaseConnector>(
					ConnectorFactory.getInstance(), 1, GenericKeyedObjectPool.WHEN_EXHAUSTED_FAIL, GenericKeyedObjectPool.DEFAULT_MAX_WAIT);
			internal_pool = factory.createPool();
		}
		return pool;
	}
	
	@Override
	public BaseConnector borrowObject(CommonConnPara connPara) {
		try {
			logger.debug("ReceiveConnectorPool.borrowObject - key:"+getConnKey(connPara.getTransportType(), connPara.getTransportPara()));
			BaseConnector connector = internal_pool.borrowObject(getConnKey(connPara.getTransportType(), connPara.getTransportPara()));
			if (connector != null) {
				connector.setEncoding(connPara.getEncoding());
			}
			return connector;
		} catch (NoSuchElementException ex) {
			logger.warn("No available connector in pool - " + connPara + ": " + ex.getMessage());
			return null;
		} catch (IllegalStateException ex) {
			logger.warn("ReceiveConnectorPool in illegal status", ex);
			return null;
		} catch (Exception ex) {
			logger.warn("makeObject error", ex);
			return null;
		}
	}
	
//	@Override
//	public void returnObject(String connType, String connPara, BaseConnector connector) {
//		try {
//			internal_pool.returnObject(getConnKey(connType, connPara), connector);
////			logger.debug("ReceiveConnectorPool.returnObject - " + internal_pool.getNumIdle());
//		} catch (Exception ex) {
//			logger.debug("returnObject error", ex);
//		}
//	}
	
	public void returnObject(BaseConnector connector) {
		try {
			internal_pool.returnObject(getConnKey(connector.getConnType(), connector.getConnPara()), connector);
			logger.debug("ReceiveConnectorPool.returnObject - " + getConnKey(connector.getConnType(), connector.getConnPara()));
		} catch (Exception ex) {
			logger.debug("returnObject error", ex);
		}
	}
}
